package publisherconfirms;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import utils.RabbitMQUtil;

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

public class ConfirmMessage {

    // 批量发消息的数量
    public static final int MESSAGE_COUNT = 1000;

    public static void main(String[] args) throws Exception {

        publishMessageIndividually(); // 989ms
        //publishBatch(); // 108ms
        //publishMessageAsync(); // 40ms


    }

    // 单个消息发布确认
    public static void publishMessageIndividually() throws Exception {
        // 得到一个Channel
        Channel channel = RabbitMQUtil.getChannel();
        // 队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, true, null);
        // 开启发布确认
        channel.confirmSelect();
        // 开始时间
        long beginTime = System.currentTimeMillis();
        //批量发消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            // 单个消息就马上进行发布确认
            boolean flag = channel.waitForConfirms(5000);
            if (flag) {
                System.out.println("消息发送成功"+message);
            }
        }
        long endTime = System.currentTimeMillis();
        System.nanoTime();
        System.out.println("发布" + MESSAGE_COUNT + "个单独确认消息，耗时" + (endTime - beginTime) + "ms");
    }

    /**
     * 批量发布确认
     *
     * @throws Exception exception
     */
    public static void publishBatch() throws Exception {
        Channel channel = RabbitMQUtil.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, true, null);
        // 开启发布确认
        channel.confirmSelect();
        long beginTime = System.currentTimeMillis();
        // 批量确认消息的大小
        int batchSize = 100;
        // 批量发消息，批量发布确认
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
            if (i % batchSize == 0) {
                channel.waitForConfirms();
            }
        }
        long endTime = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "条批量确认的消息，耗时" + (endTime - beginTime) + "ms");
    }

    // 异步确认发布
    public static void publishMessageAsync() throws Exception {
        
        Channel channel = RabbitMQUtil.getChannel();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, true, false, false, null);
        channel.confirmSelect();
        
        // 线程安全有序的哈希表，适用于高并发的情况
        ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
        // 消息确认成功，回调函数
        ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
            if (multiple) {
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag,true);
                confirmed.clear();
            } else {
                System.out.println("发送并且确认的消息系列号："+deliveryTag+" 消息体: "+outstandingConfirms.get(deliveryTag));
                outstandingConfirms.remove(deliveryTag);
            }
        };
        // 消息确认失败，回调函数
        ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
            String message = outstandingConfirms.get(deliveryTag);
            System.out.println("未确认的消息是：" + message + "未确认消息的tag：" + deliveryTag);
            // 需要使用确认回调函数清楚map集合中的未完成的确认
            ackCallback.handle(deliveryTag, multiple);
        };
        // 在channel上注册消息监听器，监听哪些消息成功了，哪些消息失败了（这里会开启一个新的线程）
        channel.addConfirmListener(ackCallback, nackCallback);
        
        // 发送消息(发送消息使用原来的那个线程)
        long beginTime = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            // 记录所有发送的消息(记录要发布的序列号要写在发布之前)
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", queueName, null, message.getBytes());

        }
        long endTime = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "个异步发布确认消息，耗时" + (endTime - beginTime) + "ms");
    }
}
